package com.pangyangyang.elasticsearch.plugin.collector;

import com.pangyangyang.elasticsearch.plugin.canal.collector.CanalCollector;
import com.pangyangyang.elasticsearch.plugin.collector.EventCollector;
import com.pangyangyang.elasticsearch.plugin.collector.EventCollectorFactory;
import com.pangyangyang.elasticsearch.plugin.river.Build;
import com.pangyangyang.elasticsearch.plugin.river.MessageRiverSettings;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;
import org.elasticsearch.common.settings.Settings;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 管理Canal的客户端链接
 * Created by yangwang on 2015/1/15.
 */
public class SingleCollectorManager {

    private static Map<String, CollectorReference> collectorMap = new HashMap<String, CollectorReference>();
    private static ESLogger logger = ESLoggerFactory.getLogger(Build.LOGGER_NAME);

    /**
     *  建立连接到Canal的Collector。
     *  确保一个Canal只会存在一个CanalCollector，重复的collector会被合并。
     *  关闭collector会考虑引用计数，只有引用计数为0才会关闭collector。<br>
     *   synchronized 在ES重新启动后会start所有plugin，这里为了避免并发导致一个collector被重复创建
     * @param settings 包括host、port、username、password信息
     */
    public static synchronized EventCollector startCanalCollector(MessageRiverSettings settings, String collectorName){
        logger.debug("prepare to start up collector {}.", collectorName);
        CanalCollector canalCollector = EventCollectorFactory.createAndStartCollector(settings, collectorName);
        String key = canalCollector.getResourceDigest();
        if(collectorMap.containsKey(key)){
            logger.debug("increment reference count of collector {}.", collectorName);
            collectorMap.get(key).incrementAndGet();
        } else {
            logger.debug("starting up collector {}.", collectorName);
            canalCollector.start();
            CollectorReference collectorReference = new CollectorReference(canalCollector);
            collectorReference.incrementAndGet();
            collectorMap.put(key, collectorReference);
        }

        return collectorMap.get(key).getCollector();
    }

    public static void stopCanalCollector(EventCollector collector){
        logger.debug("decrement reference count of collector {}.", collector.getClass().getName());
        CollectorReference collectorReference = collectorMap.get(collector.getResourceDigest());
        if(collectorReference.decrementAndGet() == 0){
            logger.debug("stopping collector {}.", collector.getClass().getName());
            collectorMap.remove(collector.getResourceDigest());
            collector.stop();
        }
    }

    protected static class CollectorReference {
        private AtomicInteger referenceCount;
        private CanalCollector collector;

        public CollectorReference(CanalCollector collector) {
            this.referenceCount = new AtomicInteger(0);
            this.collector = collector;
        }

        public int incrementAndGet(){
            return this.referenceCount.incrementAndGet();
        }

        public int decrementAndGet(){
            return this.referenceCount.decrementAndGet();
        }

        public EventCollector getCollector(){
            return this.collector;
        }
    }
}
